package org.devside.grails.queue;

import org.apache.log4j.Logger;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 异步消息队列,这是一个抽象类,客户需要实现consume方法,来决定如何消费消息
 *
 * @author matianyi@legendapl.com.cn
 */
public abstract class MessageQueue<T> implements IMessageQueue<T>, Runnable {

    private BlockingQueue<T> queue; // 消息队列

    private static final Logger LOG = Logger.getLogger(MessageQueue.class);

    private QueueStatus status = QueueStatus.STOPED; // 状态

    private ThreadGate consumeGate = new ThreadGate();  // 消费队列的门
    private ThreadGate queueGate = new ThreadGate(); // 添加消息的门
    private ThreadGate stopGate = new ThreadGate(); // 关闭队列所用的门
    private Thread thread;

    /**
     * 开始队列处理, 初始化状态
     */
    public synchronized void init() {
        if (!isRunning()) {
            int bound = getQueueBound();
            if (bound <= 0) {
                bound = Integer.MAX_VALUE;
            }
            queue = new LinkedBlockingQueue<T>(bound);
            status = QueueStatus.RUNNING;
            queueGate.open();
            consumeGate.open();
            stopGate.close(); // 关闭队列的门, 开始是关闭的, 当队列关闭后, 这个门打开

            thread = new Thread(this);
            thread.start();
            LOG.info(getClass().getName() + " started with bound " + bound);
        } else {
            LOG.info("you try to initialize a running queue : " + getClass().getName());
        }
    }

    /**
     * 获得队列的边界, 总共队列中能放多少消息
     *
     * @return 边界值
     */
    protected abstract int getQueueBound();


    /**
     * 消费消息
     *
     * @param msg 消息
     */
    protected abstract void consume(T msg);

    /*
     * (non-Javadoc)
     * 
     * @see com.legendapl.common.concurrent.IMessageQueue#run()
     */
    public void run() {
        while (isRunning()) {
            try {
                consumeGate.tryPass(); // 尝试消费门
            } catch (InterruptedException e) {
                thread.interrupt();
            }

            // 如果强制停止
            if (status == QueueStatus.BREAKING) {
                status = QueueStatus.STOPED;
                break;
            }

            // 消费消息
            T msg = null;
            try {
                msg = queue.poll(5, TimeUnit.SECONDS); // 尝试5秒钟取数据, 这里不能一直等待,否则会关闭不掉
            } catch (InterruptedException e) {
                thread.interrupt();
            }

            // 如果取到了消息, 消费之
            if (msg != null) {
                try {
                    // 消费消息快
                    consume(msg);
                } catch (Exception e) {
                    // 对消费消息的异常, 只能看着了, 最好在消费消息时自行处理
                    if (LOG.isDebugEnabled()) {
                        LOG.debug(e.getMessage(), e);
                    }
                }
            }

            // 如果正在停止, 并且队列中没有消息了
            if (status == QueueStatus.STOPING && queue.size() == 0) {
                status = QueueStatus.STOPED; // 设置停止状态, 现在已经完成停止.
                break; // 跳出循环
            }


        }
        status = QueueStatus.STOPED; // 停止运行了
        stopGate.open();
        LOG.info(getClass().getName() + " stoped. ");
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.legendapl.common.concurrent.IMessageQueue#isRunning()
     */
    public boolean isRunning() {
        return status != QueueStatus.STOPED;
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.legendapl.common.concurrent.IMessageQueue#addMessage(com.legendapl.common.concurrent.MessageBlock)
     */
    public synchronized void addMessage(T msg) {
        try {
            queueGate.tryPass(); // 尝试通过门, 如果当前队列处于挂起(HANGUP)状态, 停在这里
        } catch (InterruptedException e) {
            thread.interrupt();
        }

        if (isRunning()) {
            this.queue.add(msg);
        }
    }


    public synchronized void close() {
        sendCommand(QueueCommand.STOP);
        try {
            stopGate.tryPass();
        } catch (InterruptedException e) {
            thread.interrupt();
        }
    }

    public synchronized void closeAndFlush() {
        sendCommand(QueueCommand.BREAK);
        try {
            stopGate.tryPass();
        } catch (InterruptedException e) {
            thread.interrupt();
        }
    }

    public synchronized void hangup() {
        sendCommand(QueueCommand.HANGUP);
    }

    public void resume() {
        sendCommand(QueueCommand.RESUME);
    }

    /**
     * @see IMessageQueue#sendCommand(QueueCommand)
     */
    public void sendCommand(QueueCommand cmd) {
        switch (cmd) {
            case STOP: // 停止接受消息, 处理队列中剩余消息
                if (isRunning()) {
                    queueGate.close(); // 关闭添加消息到列队的门, 不能再添加消息了
                    status = QueueStatus.STOPING; // 标识正在停止状态
                }
                break;
            case BREAK: // 停止接受消息, 并废弃地处理队列中剩余消息
                if (isRunning()) {
                    queueGate.close();
                    status = QueueStatus.BREAKING;
                }
                break;
            case HANGUP: // 停止接受消息, 并停止处理消息, 但是保留废弃地处理队列中剩余消息
                if (isRunning()) {
                    queueGate.close();  // 将添加消息的门关闭
                    consumeGate.close(); // 将消费消息的门关闭
                    status = QueueStatus.PAUSING;
                }
                break;
            case RESTART: // 重启列队, 当队列已经STOP或BREAK后有效
                if (!isRunning()) {
                    init();
                }
                break;
            case RESUME: // 当队列已经被HANGUP后, 这个命令恢复队列
                if (status == QueueStatus.PAUSING) {
                    queueGate.open(); // 将添加消息的门打开
                    consumeGate.open(); // 将消费消息的门打开
                    status = QueueStatus.RUNNING;
                }
                break;
        }
    }
}
